
import datetime
import json
import os
import subprocess
from base64 import b64decode
from typing import Any, Union, List, Optional, Tuple, Dict
import allure
import minium
from allure_commons.types import AttachmentType
from appium.webdriver.webdriver import WebDriver
from httpx import Response
from minium import WXMinium, Page
from minium.miniprogram.base_driver.element import BaseElementType
from selenium import webdriver as web_driver
from appium import webdriver as app_driver
from openpyxl.styles import PatternFill
from selenium.common.exceptions import WebDriverException
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from case.execlHandle import ExcelHandle
from errors.platformError import ResponseException
from handle.AppletDependentHandle import AppletDependentHandle
from handle.apiDependent import ApiDependentHandle
from handle.uiDependent import UiDependentHandle
from requestProcessing.httpRequest import HttpRequest
from appium.webdriver import WebElement as AppElement
from selenium.webdriver.remote.webelement import WebElement
from utils.decorator import write_back
from variable import dependent_data
from random import randint
import time
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from utils import *
from selenium.webdriver import ActionChains
import ast
import re
from string import Template

class ApiService:

    def __init__(self, filepath: str = "dataSource/full_stack_test.xlsx"):
        self.filepath = filepath
        self.adh = ApiDependentHandle()
        self.hr = HttpRequest()
        self.eh = ExcelHandle(self.filepath)
        self.udh = UiDependentHandle()

        self.engine = None
        self.session = None
        self.driver = None
        self.wxminium = None
        self.is_recording = False

        # self.record_time = 60
        # self.video_path = "./video"
        # self.file_name = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
        # self.start_cmd = "adb shell screenrecord --time-limit {record_time} /data/local/tmp/{name}.mp4".format(
        #     record_time=self.record_time, name=self.file_name)
        # self.pull_cmd = "adb pull /data/local/tmp/{name}.mp4 {video_path}".format(name=self.file_name,
        #                                                                           video_path=self.video_path)
        # self.clear_cmd = "adb shell rm /data/local/tmp/{name}.mp4".format(name=self.file_name)
        # self.video_proc = None

    # @write_back
    def get_database_session(self, database_info: str):
        if database_info != "" and database_info is not None:
            self.report_details("数据库配置信息", database_info)

            if self.session is not None:
                self.engine.dispose()
                self.session.close()

            # database_info = self.adh.matching_variable(database_info)
            database_info = self.adh.new_matching_variable(database_info)
            self.engine = create_engine(
                "{}+{}://{}:{}@{}:{}/{}".format(database_info["type"], database_info["driven"],
                                                database_info["username"], database_info["password"],
                                                database_info["host"],
                                                database_info["port"], database_info["dbname"]),
                max_overflow=10)
            self.session = sessionmaker(bind=self.engine)()

    def playback(self):
        if self.is_recording and (isinstance(self.driver, WebDriver) or isinstance(self.driver, Page)):
            video: bytes = self.driver.stop_recording_screen()
            self.report_details("回放", b64decode(video), allure.attachment_type.MP4)

            # self.video_proc.terminate()
            # self.video_proc.send_signal(signal.CTRL_C_EVENT)
            # self.video_proc.wait()
            # subprocess.Popen(self.pull_cmd, shell=True).wait()
            # allure.attach.file(self.video_path + "/" + self.file_name + ".mp4", name="adb录屏",
            #                    attachment_type=allure.attachment_type.MP4)
            # subprocess.Popen(self.clear_cmd, shell=True).wait()

            # os.system("taskkill /IM scrcpy.exe")
            # self.video_proc.wait()
            # allure.attach.file(self.video_path + "/" + self.file_name + ".mp4", name="adb录屏",
            #                    attachment_type=allure.attachment_type.MP4)

            self.is_recording = False

    # @write_back
    def get_ui_driver(self, driver_info: str, element_info: str):
        if driver_info != "" and driver_info is not None:
            self.report_details("驱动信息", driver_info)

            if self.driver is not None:
                self.playback()

                if isinstance(self.driver, Page):
                    self.wxminium.shutdown()
                    self.wxminium = None
                else:
                    self.driver.quit()
                    self.driver = None

            # driver_info = self.adh.matching_variable(driver_info)
            driver_info = self.adh.new_matching_variable(driver_info)
            element = self.adh.matching_variable(element_info)
            for k, v in driver_info.items():
                if k == "app":
                    if isinstance(v, tuple):
                        self.driver = app_driver.Remote(element, v[0])
                        getattr(self.udh, v[1])(self.driver, v[2])
                    else:
                        self.driver = app_driver.Remote(element, v)

                elif k == "applet":
                    self.wxminium: WXMinium = minium.Minium(v)
                    self.driver = self.wxminium.app.redirect_to(element)

                    self.udh = AppletDependentHandle()

                else:
                    options = None
                    if isinstance(k, tuple):
                        if isinstance(v, tuple):
                            options = getattr(web_driver, v[0]+"Options")()
                        else:
                            options = getattr(web_driver, v+"Options")()

                        options.add_experimental_option('mobileEmulation', {'deviceName': k[2]})
                        try:
                            if k[3] != "":
                                setattr(options, "binary_location", k[3])
                        except IndexError:
                            pass

                        for argument in k[4:]:
                            if argument != "":
                                getattr(options, "add_argument")(argument)

                    if isinstance(v, tuple):
                        browser_options = {
                            v[0].lower()+"_options": options
                        }
                        if len(v) > 3:
                            self.driver = getattr(web_driver, v[3][0])(command_executor=v[3][1], options=options)
                        else:
                            self.driver = getattr(web_driver, v[0])(**browser_options)
                        getattr(self.udh, v[1])(self.driver, v[2])
                    else:
                        browser_options = {
                            v.lower()+"_options": options
                        }
                        self.driver = getattr(web_driver, v)(**browser_options)

                    self.driver.maximize_window()
                    self.driver.get(element)

    def analysis_files(self, file_info: str):
        if file_info != "" and file_info is not None:
            # for k, v in self.adh.matching_variable(file_info).items():
            for k, v in self.adh.new_matching_variable(file_info).items():
                if isinstance(v, list):
                    files = []
                    for path in v:
                        files.append((k, (open(path, 'rb'))))
                else:
                    files = {k: open(v, 'rb')}

            # file_list = file_path.split(";")
            # files = []
            # if len(file_list) > 1:
            #     for path in file_list:
            #         files.append(('files', (open(path, 'rb'))))
            # else:
            #     files = {'file':open(file_list[0],'rb')}

            return files

    # @write_back
    def operation_element(self, driver_info: str, element_info: str, extract_var: str, var_name: str,
                          need_data: str) -> Union[
        WebElement, AppElement, BaseElementType]:
        if (driver_info == "" or driver_info is None) and element_info != "" and element_info is not None:
            if isinstance(self.driver, Page):
                self.driver = self.wxminium.app.current_page

            if self.is_recording is False and (isinstance(self.driver, WebDriver) or isinstance(self.driver, Page)):
                self.driver.start_recording_screen()

                # self.video_proc = subprocess.Popen(self.start_cmd, shell=True)

                # self.video_proc = subprocess.Popen("scrcpy -r " + self.video_path + "/" + self.file_name + ".mp4",
                #                                    shell=True)

                self.is_recording = True

            # element_info = self.adh.matching_variable(element_info)
            element_info = self.adh.new_matching_variable(element_info)
            self.report_details("元素定位", str(element_info))
            element = None
            for k, v in element_info.items():
                if isinstance(k, tuple):
                    if len(k) < 3:
                        element = self.udh.get_element_all(self.driver, *k)
                    elif len(k) == 3 and type(k[2]) is not tuple:
                        if k[2] == 0:
                            element_list = self.udh.get_elements_all(self.driver, *k[:2])
                            element = element_list[randint(0, len(element_list) - 1)]
                        else:
                            element = self.udh.get_elements_all(self.driver, *k[:2])[0]

                    elif len(k) == 3 and type(k[2]) is tuple:
                        element = self.udh.display_wait(self.driver, *k)
                    else:
                        if k[3] == 0:
                            element_list = self.udh.displays_wait(self.driver, *k)
                            element = element_list[randint(0, len(element_list) - 1)]
                        else:
                            element = self.udh.displays_wait(self.driver, *k)[0]

                    if extract_var != "" and extract_var is not None:
                        extract_var = ast.literal_eval(extract_var)
                        for y, u in extract_var.items():
                            dependent_data[y] = self.adh.get_display_info(u, element)

                    if var_name != "" and var_name is not None and need_data != "" and need_data is not None:
                        var_name_list = var_name.split(";")
                        need_data_list = need_data.split(";")
                        for i in range(len(var_name_list)):
                            dependent_data[var_name_list[i]] = self.adh.get_display_info(need_data_list[i], element)

                if isinstance(v, tuple):
                    if isinstance(v[1], tuple) or isinstance(v[1], list):
                        getattr(self.udh, v[0])(self.driver, *v[1])
                    else:
                        if v[0] == "js_special_handling":
                            getattr(self.udh, v[0])(self.driver, element, v[1])
                        else:
                            getattr(self.udh, v[0])(element, v[1])
                else:
                    if v != "":
                        getattr(self.udh, v)(element)

            return element

    def update_whether(self, current_info: dict, result: Any, whether: str, color: str):
        if result:
            self.eh.wb[current_info["sheet"]].cell(row=current_info["row"], column=24,
                                                       value=repr(result))

        self.eh.wb[current_info["sheet"]].cell(row=current_info["row"], column=25,
                                                   value=whether).fill = PatternFill(fill_type='solid',
                                                                                     fgColor=color)

    def save_excel(self):
        self.eh.wb.save("./" + self.filepath)

    def update_last_error(self, assert_info: str, expected_results: str, actual_results: str, current_info: dict):
        if (assert_info is None or assert_info == "") and (expected_results is None or expected_results == "") and (
                actual_results is not None and actual_results != ""):
            self.eh.wb[current_info["sheet"]].cell(row=current_info["row"], column=24,
                                                   value="")

    def screenshot(self, element: Union[WebElement, AppElement, BaseElementType] = None, element_info: str = ""):
        if element is not None or (element_info != "" and element_info is not None):
            if isinstance(self.driver, Page):
                method = self.wxminium.app.screen_shot
            else:
                method = self.driver.get_screenshot_as_png

            self.report_details("失败截图", method(), allure.attachment_type.PNG)

    # def update_execl(self, assert_info: str, expected_results: str, actual_result_template: str, actual_results: str,
    #                  result, current_info: dict,
    #                  ident: dict):
    def update_execl(self, assert_info: str, expected_results: str, actual_result_template: str,
                     actual_results: str, result, current_info: dict,
                     resp: Union[Response, List], element: Union[WebElement, AppElement, BaseElementType]):
        initial = result
        try:
            # if expected_results != "" and expected_results is not None and ident["end"] is False:
            if expected_results != "" and expected_results is not None:
                self.adh.recursive_search_key(initial, True)

            if assert_info != "" and assert_info is not None:
                result = {}
                self.report_details("断言(预期:实际)", self.adh.result_assertion_by_dict(assert_info, resp, element, result))

        except AssertionError:
            self.screenshot(element)

            whether = "fail"
            color = "FFC125"

        else:
            whether = "pass"
            color = "AACF91"

        if (assert_info == "" or assert_info is None) and expected_results != "" and expected_results is not None:
            # result = self.adh.matching_variable(actual_result_template)
            result = self.adh.new_matching_variable(actual_result_template)
            self.report_details("断言", {
                "预期": initial,
                "实际": result
            })

        self.update_whether(current_info, result, whether, color)

        self.update_last_error(assert_info, expected_results, actual_results, current_info)

        self.save_excel()

        if whether == "fail":
            raise AssertionError

    def report_details(self, details: str, specific: Union[str, Dict[str, Any], bytes] = None,
                       attachment_type: AttachmentType = allure.attachment_type.JSON):
        with allure.step(details):
            if specific:
                if isinstance(specific, Dict):
                    specific = json.dumps(specific, ensure_ascii=False, indent=4)

                allure.attach(specific, "详情", attachment_type)

    def implement_case(self, id: str, title: str, enable: str, user_var: str, before_handle: str, method: str,
                       database_info: str,
                       driver_info: str,
                       header: str, url_or_sql: str, element_info: str,
                       param_type: str, params: str, file_info: str, extract_var: str, var_name: str,
                       need_data: str,
                       random_info: str, forced_type: str, after_handle: str, assert_info: str,
                       expected_results: str, actual_result_template: str, actual_results: str, whether: str,
                       current_info: dict = None,
                       ident: Optional[dict] = None) -> Tuple[dict, Response, Union[WebElement, AppElement, BaseElementType]]:

        if enable == "FALSE" or enable == "Status.DISABLE":
            return

        if user_var != "" and user_var is not None:
            # user_var = self.adh.matching_variable(user_var)
            user_var = self.adh.new_matching_variable(user_var)
            self.report_details("用户变量", user_var)
            for k, v in user_var.items():
                dependent_data[k] = v

        if before_handle != "" and before_handle is not None:
            before_script_map: dict = {
                "参数替换前": before_handle
            }
            before_handle = self.adh.matching_variable(before_handle)
            before_script_map["参数替换后"] = before_handle
            exec(before_handle)
            self.report_details("前置处理脚本", before_script_map)

        result = None
        if expected_results != "" and expected_results is not None:
            # result = self.adh.matching_variable(expected_results)
            result = self.adh.new_matching_variable(expected_results)

        self.get_database_session(database_info)

        self.get_ui_driver(driver_info, element_info)

        resp = None
        if (method == "" or method is None) and url_or_sql != "" and url_or_sql is not None:
            self.playback()

            url_or_sql = self.adh.matching_variable(url_or_sql)
            self.report_details("sql语句", url_or_sql)

            if "select" in url_or_sql or "SELECT" in url_or_sql:
                resp = self.session.execute(url_or_sql).fetchall()
                self.report_details("执行sql结果", str(resp))

            else:
                self.session.execute(url_or_sql)

            self.session.commit()

            if "select" in url_or_sql or "SELECT" in url_or_sql:
                self.report_details("执行sql提取参数(dict)", self.adh.dependent_data_acquisition_by_dict(extract_var, resp))

                self.report_details("执行sql提取参数(tuple)", self.adh.dependent_data_acquisition(var_name, need_data, random_info,
                                                                                forced_type, resp))

        files = None
        if file_info != "" and file_info is not None:
            files = self.analysis_files(file_info)
            self.report_details("文件信息", str(files))

        if method != "" and method is not None:
            self.playback()

            self.report_details("请求方式", method)

            if param_type == "from" or param_type == "json":
                self.report_details("参数类型", param_type)

                # params = self.adh.dependent_data_assignment(params, False)   # 旧的替换参数
                # params = self.adh.matching_variable(params)
                params = self.adh.new_matching_variable(params)
                self.report_details("请求参数", params)

            url_or_sql = self.adh.matching_variable(url_or_sql)
            self.report_details("请求地址", url_or_sql)

            if header != "" and header is not None:
                # header = self.adh.dependent_data_assignment(params, False)   # 旧的替换参数
                # header = self.adh.matching_variable(header)
                header = self.adh.new_matching_variable(header)
                self.report_details("请求头信息", header)

            resp = self.hr.request(method, url_or_sql, header, param_type, params, files)
            self.report_details("请求响应结果", resp.json())
            if resp.status_code != 200:
                raise ResponseException(resp.text)
            else:
                self.report_details("请求接口提取参数(dict)", self.adh.dependent_data_acquisition_by_dict(extract_var, resp))

                self.report_details("请求接口提取参数(tuple)", self.adh.dependent_data_acquisition(var_name, need_data, random_info,
                                                                               forced_type, resp))

        # self.operation_element(current_info, ident, driver_info, element_info, extract_var, var_name, need_data)
        element = self.operation_element(driver_info, element_info, extract_var, var_name, need_data)

        if after_handle != "" and after_handle is not None:
            after_script_map: dict = {
                "参数替换前": after_handle
            }
            after_handle = self.adh.matching_variable(after_handle)
            after_script_map["参数替换后"] = after_handle
            exec(after_handle)
            self.report_details("后置处理脚本", after_script_map)

        return result, resp, element

    def run_api(self, case: dict):
        # case["ident"] = {
        #     "end":False
        # }

        try:
            result, resp, element = self.implement_case(**case)

        except Exception as e:
            if type(e) is not WebDriverException and type(e) is not AttributeError:
                self.screenshot(element_info=case["element_info"])

            self.update_whether(case["current_info"], e, "fail", "FFC125")
            self.save_excel()

            # raise AssertionError
            raise e

        else:
            print("dependent_data：", dependent_data)

            # self.update_execl(case["assert_info"], case["expected_results"], case["actual_result_template"],
            #                   case["actual_results"], result,
            #                   case["current_info"],
            #                   case["ident"])
            self.update_execl(case["assert_info"], case["expected_results"], case["actual_result_template"],
                              case["actual_results"], result,
                              case["current_info"], resp,
                              element)












































